package a_o2odata_deal.utils.zhibo_qy

import a_aa_amainpackage.a_o2odata_deal.config.config._
import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions.lit

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/2/11 16:42
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_all_anchor {
  def handle_all_anchor(sqlContext:SQLContext):Unit={
    val data_id = sqlContext.read.json(s"${a_aa_amainpackage.a_o2odata_deal.config.config.tmall_live_id_path}")

    data_id.createOrReplaceTempView("data_id")

    //计算每个主播的场次,正式代码
    sqlContext.sql(
      s"""
         |select
         |anchorId,
         |anchorFans
         |from
         |(select
         |anchorId,
         |anchorFans,
         |row_number() over(partition by anchorId order by cast(liveTime as bigint) desc) as rank
         |from data_id)
         |where rank=1
       """.stripMargin).createOrReplaceTempView("data1")

    sqlContext.sql(
      s"""
         |select
         |anchorId,
         |count(distinct liveId) as liveCount,
         |sum(viewerCount) as viewerCount,
         |--sum(anchorFans) as anchorFans,
         |sum(commodityCount) as liveGoodCount
         |from data_id
         |group by anchorId
       """.stripMargin).createOrReplaceTempView("data2")

    val result = sqlContext.sql(
      s"""
         |select
         |t2.*,
         |t1.anchorFans
         |from data1 t1
         |inner join
         |data2 t2
         |on t1.anchorId=t2.anchorId
       """.stripMargin)

    //sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/tmall/all_userid/*/*")
    //  .selectExpr("userId").dropDuplicates().createOrReplaceTempView("tm_userid")
    //sqlContext.read.json("s3a://o2o-dataproces-group/panzonghao/taobao/all_userid/*/*")
    //  .selectExpr("userId").dropDuplicates().createOrReplaceTempView("tb_userid")
    sqlContext.read.json(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/backups/all_userid/*/*")
      .dropDuplicates("userId")
      .createOrReplaceTempView("alluserid")
    result.createOrReplaceTempView("all_data")

    val result1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.userId is null then '' else platForm end as from_where
         |from all_data t1
         |left join
         |alluserid t2
         |on t1.anchorId=t2.userId
       """.stripMargin)
    /*sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.userId is null then '' else 'tmall' end as from_where
         |from all_data t1
         |left join
         |tm_userid t2
         |on t1.anchorId=t2.userId
         """.stripMargin).createOrReplaceTempView("mid")

    val result1 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.userId is null then t1.from_where else 'taobao' end as from_wheres
         |from mid t1
         |left join
         |tb_userid t2
         |on t1.anchorId=t2.userId
         """.stripMargin).drop("from_where")
      .withColumnRenamed("from_wheres","from_where")*/

    val v = sqlContext.read.json(s"${tmall_anchor_id_path}").toJSON.rdd.map(line =>{
      val nObject: JSONObject = JSON.parseObject(line)
      val nick: String = nObject.getOrDefault("nick", "-1").toString()
      nObject.put("nick",nick)
      nObject.toString
    })


    val anchor_id = sqlContext.read.json(v).where("nick!='-1'").selectExpr("anchorId","nick").dropDuplicates("anchorId")
    anchor_id.createOrReplaceTempView("anchor_id")


    result1.createOrReplaceTempView("source_data")

    val result2 = sqlContext.sql(
      s"""
         |select
         |t1.*,
         |case when t2.anchorId is null then '-1' else t2.nick end as nick
         |from source_data t1
         |left join
         |anchor_id t2
         |on t1.anchorId=t2.anchorId
         """.stripMargin)

    //result2.createOrReplaceTempView("result2")
//
    //sqlContext.sql(
    //  s"""
    //     |select
    //     |t1.*,
    //     |case when t2.userId is null then '' else 'tmall' end as from_where
    //     |from result2 t1
    //     |left join
    //     |tm_userid t2
    //     |on t1.anchorId=t2.userId
    //     """.stripMargin).createOrReplaceTempView("mid")
//
    //val result3 = sqlContext.sql(
    //  s"""
    //     |select
    //     |t1.*,
    //     |case when t2.userId is null then t1.from_where else 'taobao' end as from_wheres
    //     |from mid t1
    //     |left join
    //     |tb_userid t2
    //     |on t1.anchorId=t2.userId
    //     """.stripMargin).drop("from_where")
    //  .withColumnRenamed("from_wheres","from_where")

    result2.withColumn("timeStamp",lit(timeStamp))
      .repartition(4).write.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/all_anchor")
  }
}
